package com.atguigu.transform;

import com.atguigu.pojo.AppEvent;
import com.atguigu.pojo.Event;
import com.atguigu.pojo.ThirdPartyEvent;
import com.sun.jersey.core.impl.provider.entity.XMLRootObjectProvider;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/14 20:02
 *
 * 对账案例
 *      我们可以实现一个实时对账的需求，也就是app支付操作和第三方支付操作的一个双流Join。
 *      目前只关心对账成功的数据不限制两方数据的延迟
 *
 *
 */
public class Flink11_BillCheck {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //基于端口模拟数据
        //AppEvent
        //order-1,create_1000
        //order-1,pay,2000
        SingleOutputStreamOperator<AppEvent> appDs = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new AppEvent(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }
                ).filter(
                        event -> "pay".equals(event.getEventType())
                );

        // ThirdpartyEvent
        // order-1,pay,wechat,2000

        SingleOutputStreamOperator<ThirdPartyEvent> thirdDs = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new ThirdPartyEvent(fields[0].trim(), fields[1].trim(), fields[2].trim(), Long.valueOf(fields[3].trim()));


                        }
                );


        //connect
        SingleOutputStreamOperator<String> processDs = appDs.connect(thirdDs)
                .process(
                        new CoProcessFunction<AppEvent, ThirdPartyEvent, String>() {
                            // 使用集合来维护遭到的数据

                            //自己维护状态
                            private Map<String, AppEvent> appEventMap = new HashMap<String, AppEvent>();
                            private Map<String, ThirdPartyEvent> thirdPartyEventHashMap = new HashMap<String, ThirdPartyEvent>();

//                            List<AppEvent> appEventList = new ArrayList<AppEvent>();
//                            List<ThirdPartyEvent> thirdPartyEvents = new ArrayList<ThirdPartyEvent>();


                            //处理App Event
                            @Override
                            public void processElement1(AppEvent value, CoProcessFunction<AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> out) throws Exception {

                                // 1. 先到对方的状态中看看是否又对应的数据
                                if (thirdPartyEventHashMap.containsKey(value.getOrderId())) {
                                    out.collect(value.getOrderId() + "对账成功！！！");
                                    // 从对方的状态中清除数据
                                    thirdPartyEventHashMap.remove(value.getOrderId());
                                } else {
                                    // appEvent 到了，thridPartyEvent没到
                                    appEventMap.put(value.getOrderId(), value);
                                }
                            }

                            //处理ThirdPartEvent
                            @Override
                            public void processElement2(ThirdPartyEvent value, CoProcessFunction<AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> out) throws Exception {

                                // 1. 先到对方的状态中看看是否又对应的数据
                                if (appEventMap.containsKey(value.getOrderId())) {
                                    out.collect(value.getOrderId() + "对账成功！！！");
                                    // 从对方的状态中清除数据
                                    appEventMap.remove(value.getOrderId());
                                } else {
                                    // appEvent 到了，thridPartyEvent没到
                                    thirdPartyEventHashMap.put(value.getOrderId(), value);
                                }
                            }
                        }
                );

        processDs.print();

        /*
        appDs.keyBy(
                AppEvent::getOrderId
        ).connect(
                thirdDs.keyBy(
                        ThirdPartyEvent::getOrderId
                )
        )

         */

        env.execute();


    }
}
